package com.health.trans.processor.writer;

import com.health.trans.annotation.Writer;
import com.health.trans.kit.WriterConfig;
import com.health.trans.pojo.JobConfig;
import io.swagger.annotations.ApiModelProperty;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.kafka.KafkaItemWriter;

import java.util.List;

/**
 * @author nqlz
 */
@Slf4j
@Writer
public class KafkaWriter<K>  implements CsWriter {


    @Override
    public <T> ItemWriter<T> getWriter(WriterConfig<T> writerConfig) {
        KafkaItemWriter<K, T> kafkaItemWriter = new KafkaItemWriter<K, T>() {
            private  JobConfig<T> jobConfig;
            @Override
            public void write(List<? extends T> list) throws Exception {
                //list.parallelStream().forEach(i -> kafkaService.send(jobConfig.getWriteTableClass(), i));
                //list.parallelStream().forEach(i -> kafkaTemplate.send(jobConfig.getWriteTableClass().getSimpleName(), i));
                jobConfig.log(list.size());
            }
        };
        return kafkaItemWriter;
    }
}
